package im.redis.collection;

import im.redis.client.operations.PubSubOperation;
import im.redis.client.operations.SortedSetOperation;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.exceptions.JedisConnectionException;

/**
 * 基于Redis的分布式延迟队列
 */
public class RDelayQueue<E> {
	private final transient Logger logger = LoggerFactory.getLogger(getClass());

	private final RedisCollections collections;
	private final SortedSetOperation setOps;
	private final PubSubOperation pubSubOps;

	private final Class<E> clazz;
	private final String queueName;
	private final String pubSubName;

	private final transient ReentrantLock lock = new ReentrantLock();
	private final Bell bell = new Bell();
	private transient long nextTime = Long.MAX_VALUE;// 等待的下一个时间点，以此减少发送队列元素改变事件

	public RDelayQueue(RedisCollections collections, String keyName, Class<E> clazz) {
		super();
		this.clazz = clazz;
		this.collections = collections;
		this.pubSubOps = collections.getJedisTemplate().pubSubOps();
		this.setOps = collections.getJedisTemplate().sortedSetOps();

		this.queueName = keyName;
		this.pubSubName = "pubsub." + keyName;
	}

	public boolean offer(E e, long delayMills) {
		long score = System.currentTimeMillis() + delayMills;
		String json = collections.serialize(e);
		boolean result = setOps.zadd(queueName, score, json);
		if (result && score < nextTime) {// 发布队列元素改变事件
			Long count = pubSubOps.publish(pubSubName, json);
			if (count != null) {
				logger.debug("publish mesage {} into [{}], {} clients received", new Object[] { json, pubSubName, count });
			}
		}
		return result;
	}

	// Leader-Follower pattern
	private Thread leader = null;

	private AtomicBoolean listened = new AtomicBoolean(false);

	private QueueListener queueListener;

	public E take() throws InterruptedException {
		final ReentrantLock lock = this.lock;
		try {
			lock.lockInterruptibly();
			subcribe();
			for (;;) {
				String head = setOps.peekFirst(queueName);
				if (head == null) {
					bell.sleep();
					nextTime = Long.MAX_VALUE;
					continue;
				}
				Double score = setOps.zscore(queueName, head);
				if (score == null) {
					nextTime = Long.MAX_VALUE;
					continue;
				}
				long now = System.currentTimeMillis();
				if (score.longValue() <= now) {
					// 如果取出的跟期望的不一致，说明被其他客户端抢先获取到
					String result = setOps.expectFirst(queueName, head);
					if (result != null) {
						return collections.deserialize(result, clazz);
					}
					continue;
				}
				nextTime = now;
				head = null;
				if (leader != null) {// 已有其他线程在等待
					bell.sleep();
				} else {
					Thread thisThread = Thread.currentThread();
					leader = thisThread;
					try {
						bell.sleep(TimeUnit.MILLISECONDS, score.longValue() - now);
					} finally {
						if (leader == thisThread) {
							leader = null;
						}
					}
				}
			}
		} catch (InterruptedException e) {
			unsubscribe();
			throw e;
		} finally {
			if (leader == null) {
				bell.ring();
			}
			lock.unlock();
		}
	}

	// 监听队列元素改变事件
	private void subcribe() {
		if (!listened.getAndSet(true)) {
			new Thread(new QueueListener(), queueName + "订阅线程").start();
		}
	}

	// 取消监听
	private void unsubscribe() {
		logger.debug("occr InterruptedException, unsubscribe channel [{}]", pubSubName);
		try {
			queueListener.unsubscribe();// 发生中断异常，退出监听
		} catch (Exception e) {
		} finally {
			listened.set(false);
		}
	}

	private class QueueListener extends JedisPubSub implements Runnable {

		@Override
		public void onMessage(String channel, String message) {
			logger.debug("channel [{}] received new message {}", channel, message);
			bell.ring();
		}

		@Override
		public void run() {
			logger.debug("subcribe channel [{}] thread started", pubSubName);
			queueListener = this;
			try {
				pubSubOps.subscribe(queueListener, pubSubName);// 阻塞方法
			} catch (JedisConnectionException e) {// redis链接断开
				new Thread(new ReconnectTask(), queueName + "重连线程").start();
			}
			logger.debug("subcribe channel [{}] thread exited", pubSubName);
		}

	}

	private class ReconnectTask implements Runnable {
		@Override
		public void run() {
			while (true) {
				if (collections.getJedisTemplate().ping()) {
					new Thread(new QueueListener()).start();
					bell.ring();// 重连成功，重新唤起线程
					return;
				}

				logger.error("[{}] redis server connect failed", queueName);
				try {
					TimeUnit.SECONDS.sleep(2);
				} catch (InterruptedException e) {
					break;
				}
			}
		}
	}

	/** 每次只允许一个线程执行 */
	private class Bell {
		private final Semaphore semaphore = new Semaphore(0);

		public void sleep() throws InterruptedException {
			semaphore.acquire();
		}

		public void sleep(TimeUnit unit, long timeout) throws InterruptedException {
			semaphore.tryAcquire(timeout, unit);
		}

		public void ring() {
			if (semaphore.availablePermits() <= 0) {
				semaphore.release();
			}
		}
	}
}
